package com.xbongbong.stage.controller;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xbongbong.paas.constant.PaasConstant;
import com.xbongbong.paas.constant.TimeConstant;
import com.xbongbong.paas.elasticsearch.pojo.XbbAggregatedPage;
import com.xbongbong.paas.elasticsearch.util.EsUtil;
import com.xbongbong.paas.elasticsearch.util.XbbElasticsearchRestTemplate;
import com.xbongbong.paas.enums.DelEnum;
import com.xbongbong.paas.enums.FlowStatusEnum;
import com.xbongbong.paas.enums.IndexTypeEnum;
import com.xbongbong.paas.enums.StageTypeEnum;
import com.xbongbong.paas.redis.PaasRedisHelper;
import com.xbongbong.paas.redis.RedisPrefixConstant;
import com.xbongbong.paas.toolbox.exception.XbbException;
import com.xbongbong.paas.toolbox.util.CollectionsUtil;
import com.xbongbong.paas.toolbox.util.DateTimeUtil;
import com.xbongbong.paas.toolbox.util.StringUtil;
import com.xbongbong.paas.toolbox.wrap.BaseVO;
import com.xbongbong.paas.toolbox.wrap.XbbResponse;
import com.xbongbong.pro.constant.XbbProConstant;
import com.xbongbong.pro.enums.SaasMarkEnum;
import com.xbongbong.pro.enums.errorcodes.OpportunityErrorCodeEnum;
import com.xbongbong.pro.enums.errorcodes.SystemErrorCodeEnum;
import com.xbongbong.pro.opportunity.pojo.SaleStageSyncStatusPojo;
import com.xbongbong.pro.opportunity.pojo.dto.OpportunityStageSyncDTO;
import com.xbongbong.pro.opportunity.pojo.dto.OpportunityUpdateBatchDTO;
import com.xbongbong.pro.opportunity.pojo.dto.OpportunityUpdateDTO;
import com.xbongbong.pro.opportunity.pojo.vo.SaleStageModifyLogVO;
import com.xbongbong.pro.thread.ThreadPoolBeanConfig;
import com.xbongbong.saas.constant.ParameterConstant;
import com.xbongbong.saas.constant.StringConstant;
import com.xbongbong.saas.domain.entity.OpportunityStageLogEntity;
import com.xbongbong.saas.domain.entity.OpportunityWorkLogEntity;
import com.xbongbong.saas.domain.entity.SaleStageEntity;
import com.xbongbong.saas.domain.entity.SaleStageSyncStatusEntity;
import com.xbongbong.saas.domain.entity.StageWorkEntity;
import com.xbongbong.saas.domain.entity.ext.OpportunityEntityExt;
import com.xbongbong.saas.enums.XbbRefTypeEnum;
import com.xbongbong.saas.enums.business.SalesOpportunityEnum;
import com.xbongbong.saas.enums.dictionary.SaleStageSyncStatusEnum;
import com.xbongbong.saas.model.OpportunityStageLogModel;
import com.xbongbong.saas.model.OpportunityWorkLogModel;
import com.xbongbong.saas.model.SaleStageModel;
import com.xbongbong.saas.model.SaleStageModifyLogModel;
import com.xbongbong.saas.model.SaleStageSyncStatusModel;
import com.xbongbong.saas.model.StageWorkModel;
import com.xbongbong.saas.service.OpportunityService;
import com.xbongbong.saas.toolbox.help.FastJsonHelper;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.i18n.LocaleContextHolder;
import org.springframework.data.domain.PageRequest;
import org.springframework.http.MediaType;
import org.springframework.validation.BindingResult;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import javax.validation.Valid;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import static org.elasticsearch.index.query.QueryBuilders.boolQuery;
import static org.elasticsearch.index.query.QueryBuilders.termQuery;
import static org.elasticsearch.index.query.QueryBuilders.termsQuery;

/**
 * 编辑机会的阶段流程的阶段比例或者任务占比后同步机会主数据的阶段比例
 *
 * @author guojun.hu
 * @version v1.0
 * @date 2021/11/18 15:15
 * @since v1.0
 */
@RestController
@RequestMapping(XbbProConstant.API_ROOT_PREFIX + "/stage/sync/opportunity")
public class StageSyncOpportunityController {

    private static final Logger LOG = LoggerFactory.getLogger(StageSyncOpportunityController.class);

    @Resource
    private OpportunityService opportunityService;
    @Resource
    private SaleStageModel saleStageModel;
    @Resource
    private XbbElasticsearchRestTemplate xbbElasticsearchRestTemplate;
    @Resource
    private OpportunityStageLogModel opportunityStageLogModel;
    @Resource
    private StageWorkModel stageWorkModel;
    @Resource
    private OpportunityWorkLogModel opportunityWorkLogModel;
    @Resource
    private SaleStageModifyLogModel saleStageModifyLogModel;
    @Resource
    private PaasRedisHelper paasRedisHelper;
    @Resource
    private SaleStageSyncStatusModel saleStageSyncStatusModel;
    @Resource
    private ThreadPoolBeanConfig threadPoolBeanConfig;

    private static final int MAX_SYNC_TIMES = 10;
    private static final long MIN_EXECUTION_INTERVAL = TimeConstant.SECONDS_PER_DAY * 7;
    private static final long MAX_SYNC_DATA_COUNT = 10000L;

    /**
     * 编辑机会的阶段流程的阶段比例或者任务占比后同步机会主数据的阶段比例
     */
    @RequestMapping(value = "/stageSyncOpportunityData", method = RequestMethod.POST, produces = MediaType.APPLICATION_JSON_UTF8_VALUE)
    public String stageSyncOpportunityData(@RequestBody @Valid OpportunityStageSyncDTO stageSyncDTO, BindingResult br, @RequestParam("lang") String lang) throws Exception {
        LocaleContextHolder.setLocale(Locale.SIMPLIFIED_CHINESE);
        if (!Objects.equals(lang, Locale.SIMPLIFIED_CHINESE.toString())) {
            LocaleContextHolder.setLocale(Locale.US);
        }
        XbbResponse<BaseVO> response = new XbbResponse<>(new BaseVO());
        if (br.hasErrors()) {
            String errMsg = Objects.requireNonNull(br.getFieldError()).getDefaultMessage();
            response = new XbbResponse<>(SystemErrorCodeEnum.API_ERROR_100002, errMsg);
        } else {
            try {
                checkSync(stageSyncDTO);
            } catch (XbbException e) {
                response = new XbbResponse<>(e);
            } catch (Exception unHandledException) {
                response = new XbbResponse<>(new XbbException(SystemErrorCodeEnum.API_ERROR_100001, unHandledException.getMessage()));
            }
        }
        return JSON.toJSONString(response);
    }

    private void checkSync(OpportunityStageSyncDTO stageSyncDTO) throws XbbException {
        String corpid = stageSyncDTO.getCorpid();
        Long formId = stageSyncDTO.getFormId();
        Long stageProcessId = stageSyncDTO.getStageProcessId();
        Integer syncInProgress = stageSyncDTO.getSyncInProgress();
        Integer syncWinOrLoss = stageSyncDTO.getSyncWinOrLoss();
        if (Objects.equals(syncInProgress, 0) && Objects.equals(syncWinOrLoss, 0)) {
            throw new XbbException(OpportunityErrorCodeEnum.API_ERROR_202027);
        }

        SaleStageSyncStatusPojo saleStageSyncStatusPojo = null;
        String syncStatusKey = corpid + StringConstant.UNDER_LINE + formId + StringConstant.UNDER_LINE + stageProcessId;
        String syncStatusValue = paasRedisHelper.getValue(RedisPrefixConstant.SALE_STAGE_SYNC_STATUS, syncStatusKey);
        // 先从redis中获取同步的状态
        if (StringUtil.isNotEmpty(syncStatusValue)) {
            JSONObject syncStatusObj = FastJsonHelper.parseObject(syncStatusValue);
            saleStageSyncStatusPojo = JSON.parseObject(syncStatusObj.toJSONString(), SaleStageSyncStatusPojo.class);
        }
        // 如果redis中没有获取到同步状态，则从数据库中获取
        if (Objects.isNull(saleStageSyncStatusPojo)) {
            SaleStageSyncStatusEntity saleStageSyncStatus = saleStageSyncStatusModel.getByStageProcessId(corpid, formId, stageProcessId);
            if (Objects.nonNull(saleStageSyncStatus)) {
                saleStageSyncStatusPojo = new SaleStageSyncStatusPojo(saleStageSyncStatus.getStatus(), saleStageSyncStatus.getExecuteTime(), saleStageSyncStatus.getFinishTime(), saleStageSyncStatus.getSyncTimes());
            }
        }
        // 同步限制校验，saleStageSyncStatusPojo不为空代表之前进行过数据同步
        if (Objects.nonNull(saleStageSyncStatusPojo)) {
            Integer status = saleStageSyncStatusPojo.getStatus();
            if (Objects.equals(status, SaleStageSyncStatusEnum.PROGRESS.getCode())) {
                throw new XbbException(OpportunityErrorCodeEnum.API_ERROR_202020);
            }
            Integer syncTimes = saleStageSyncStatusPojo.getSyncTimes();
            if (syncTimes >= MAX_SYNC_TIMES) {
                throw new XbbException(OpportunityErrorCodeEnum.API_ERROR_202021);
            }
            // 获取距离上次执行同步的时间间隔
            long executeInterval = DateTimeUtil.getInt() - saleStageSyncStatusPojo.getExecuteTime();
            if (executeInterval < MIN_EXECUTION_INTERVAL) {
                throw new XbbException(OpportunityErrorCodeEnum.API_ERROR_202022);
            }
        }
        // 查询机会总数
        BoolQueryBuilder boolQueryBuilder = boolQuery();
        boolQueryBuilder.filter(termQuery(ParameterConstant.CORPID + StringConstant.POINT + StringConstant.KEY_WORD, corpid));
        boolQueryBuilder.filter(termQuery(StringConstant.DEL, DelEnum.NORMAL.getDel()));
        boolQueryBuilder.filter(termQuery(StringConstant.FORM_ID, formId));
        boolQueryBuilder.filter(termQuery(SalesOpportunityEnum.getEsAttr4Keyword(SalesOpportunityEnum.STAGE_PROCESS_ID), stageProcessId));
        boolQueryBuilder.filter(termsQuery(StringConstant.FLOW_STATUS, Arrays.asList(FlowStatusEnum.NOT_NEED_APPROVAL.getType(), FlowStatusEnum.PASS.getType())));
        // TODO 要不要包含已归档数据
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(boolQueryBuilder);
        SearchRequest searchRequest = new SearchRequest(IndexTypeEnum.IDX_SAAS_OPPORTUNITY.getIndex());
        searchRequest.source(sourceBuilder);
        CountRequest countRequest = new CountRequest(searchRequest.indices()[0]);
        countRequest.query(searchRequest.source().query());
        long totalCount = xbbElasticsearchRestTemplate.count(countRequest);
        if (totalCount > MAX_SYNC_DATA_COUNT) {
            throw new XbbException(OpportunityErrorCodeEnum.API_ERROR_202023);
        }
        if (totalCount == 0L) {
            throw new XbbException(OpportunityErrorCodeEnum.API_ERROR_202025);
        }
        // 查询tb_sale_stage_modify_log表获取编辑过的阶段流程和阶段
        SaleStageModifyLogVO saleStageModifyLogVO = saleStageModifyLogModel.getNeedSyncStageIdList(corpid, formId, stageProcessId);
        if (CollectionsUtil.isEmpty(saleStageModifyLogVO.getSaleStageModifyLogList())) {
            throw new XbbException(OpportunityErrorCodeEnum.API_ERROR_202024);
        }

        final SaleStageSyncStatusPojo saleStageSyncStatusPojoFinal = saleStageSyncStatusPojo;
        Runnable runnable = () -> {
            try {
                syncOpportunity(stageSyncDTO, saleStageModifyLogVO, saleStageSyncStatusPojoFinal, syncStatusKey);
            } catch (Exception e) {
                LOG.error("编辑机会的阶段流程的阶段比例或者任务占比后同步机会主数据的阶段比例", e);
            }
        };
        threadPoolBeanConfig.saleStageSyncThreadPool().execute(runnable);
    }

    private void syncOpportunity(OpportunityStageSyncDTO stageSyncDTO, SaleStageModifyLogVO saleStageModifyLogVO, SaleStageSyncStatusPojo saleStageSyncStatusPojo, String syncStatusKey) throws XbbException {
        String corpid = stageSyncDTO.getCorpid();
        Long formId = stageSyncDTO.getFormId();
        Long stageProcessId = stageSyncDTO.getStageProcessId();
        Integer syncInProgress = stageSyncDTO.getSyncInProgress();
        Integer syncWinOrLoss = stageSyncDTO.getSyncWinOrLoss();

        // 查询阶段
        Map<String, Object> stageParam = new HashMap<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
        stageParam.put("corpid", corpid);
        stageParam.put("del", 0);
        stageParam.put("formId", formId);
        stageParam.put("stageProcessId", stageProcessId);

        // 查询阶段任务
        Map<String, Object> stageWorkParam = new HashMap<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
        stageWorkParam.put("corpid", corpid);
        stageWorkParam.put("del", 0);
        stageWorkParam.put("businessType", XbbRefTypeEnum.SALE_STAGE.getCode());
        stageWorkParam.put("saasMark", SaasMarkEnum.SAAS.getCode());
        stageWorkParam.put("formId", formId);
        stageWorkParam.put("stageProcessId", stageProcessId);

        // 查询阶段任务记录
        Map<String, Object> workLogParam = new HashMap<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
        workLogParam.put("corpid", corpid);
        workLogParam.put("del", 0);
        workLogParam.put("isFinished", 1);

        // 用于存储已经查询出来的阶段数据。key-value：阶段id-阶段id对应的阶段实体
        Map<Long, SaleStageEntity> saleStageMap = new HashMap<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
        // 用于存储已经查询出来的阶段任务。key-value：阶段id-该阶段id对应的任务列表
        Map<Long, List<StageWorkEntity>> stageWorkMap = new HashMap<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);

        Map<String, Object> stageLogParam = new HashMap<>(PaasConstant.DEFAULT_INITIAL_CAPACITY);
        stageLogParam.put("corpid", corpid);
        stageLogParam.put("formId", formId);
        stageLogParam.put("stageProcessId", stageProcessId);
        stageLogParam.put("stageIdIn", saleStageModifyLogVO.getNeedSyncSaleStageIdList());
        stageLogParam.put("isNow", 1);
        stageLogParam.put("del", 0);
        stageLogParam.put("columns", "id, data_id");
        if (Objects.equals(syncInProgress, 1) && Objects.equals(syncWinOrLoss, 0)) {
            stageLogParam.put("stageType", StageTypeEnum.ORDINARY_STAGE.getType());
        } else if (Objects.equals(syncInProgress, 0) && Objects.equals(syncWinOrLoss, 1)) {
            stageLogParam.put("stageTypeIn", Arrays.asList(StageTypeEnum.SUCCESS_STAGE.getType(), StageTypeEnum.FAIL_STAGE.getType(), StageTypeEnum.CANCEL_STAGE.getType()));
        } else if (Objects.equals(syncInProgress, 1) && Objects.equals(syncWinOrLoss, 1)) {
            stageLogParam.put("stageTypeIn", Arrays.asList(StageTypeEnum.ORDINARY_STAGE.getType(), StageTypeEnum.SUCCESS_STAGE.getType(), StageTypeEnum.FAIL_STAGE.getType(), StageTypeEnum.CANCEL_STAGE.getType()));
        }
        List<OpportunityStageLogEntity> stageLogList = opportunityStageLogModel.findEntitys(stageLogParam);
        // 构建机会数据和当前阶段记录之间一一对应的关系
        Map<Long, Long> dataStageLogMap = stageLogList.stream().collect(Collectors.toMap(OpportunityStageLogEntity::getDataId, OpportunityStageLogEntity::getId, (v1, v2) -> v2));
        Set<Long> opportunityIdList = dataStageLogMap.keySet();

        // 用于查询需要同步的机会数据
        BoolQueryBuilder boolQueryBuilder = boolQuery();
        boolQueryBuilder.filter(termQuery(ParameterConstant.CORPID + StringConstant.POINT + StringConstant.KEY_WORD, corpid));
        boolQueryBuilder.filter(termQuery(StringConstant.DEL, DelEnum.NORMAL.getDel()));
        boolQueryBuilder.filter(termQuery(StringConstant.FORM_ID, formId));
        boolQueryBuilder.filter(termQuery(SalesOpportunityEnum.getEsAttr4Keyword(SalesOpportunityEnum.STAGE_PROCESS_ID), stageProcessId));
        boolQueryBuilder.filter(termsQuery(StringConstant.FLOW_STATUS, Arrays.asList(FlowStatusEnum.NOT_NEED_APPROVAL.getType(), FlowStatusEnum.PASS.getType())));
        boolQueryBuilder.filter(termsQuery(StringConstant.DATA_ID, opportunityIdList));
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(boolQueryBuilder);
        SearchRequest searchRequest = new SearchRequest(IndexTypeEnum.IDX_SAAS_OPPORTUNITY.getIndex());
        searchRequest.source(sourceBuilder);
        CountRequest countRequest = new CountRequest(searchRequest.indices()[0]);
        countRequest.query(searchRequest.source().query());
        long totalCount = xbbElasticsearchRestTemplate.count(countRequest);
        LOG.warn("开始同步公司{}的销售机会数据，一共{}条", corpid, totalCount);

        // 开始进行同步。更新redis【status=同步中 执行时间=当前时间 完成时间=空 同步次数+1】
        Long currentExecuteTime = DateTimeUtil.getInt();
        SaleStageSyncStatusPojo saleStageSyncStatusPojoUpdate = new SaleStageSyncStatusPojo();
        saleStageSyncStatusPojoUpdate.setStatus(SaleStageSyncStatusEnum.PROGRESS.getCode());
        saleStageSyncStatusPojoUpdate.setExecuteTime(currentExecuteTime);
        if (Objects.nonNull(saleStageSyncStatusPojo)) {
            saleStageSyncStatusPojoUpdate.setSyncTimes(saleStageSyncStatusPojo.getSyncTimes() + 1);
        } else {
            saleStageSyncStatusPojoUpdate.setSyncTimes(1);
        }
        paasRedisHelper.setValue(RedisPrefixConstant.SALE_STAGE_SYNC_STATUS, syncStatusKey, saleStageSyncStatusPojoUpdate, TimeConstant.SECONDS_PER_DAY);
        // 更新tb_sale_stage_sync_status表
        SaleStageSyncStatusEntity saleStageSyncStatusUpdate = new SaleStageSyncStatusEntity(corpid, formId, stageProcessId,
                saleStageSyncStatusPojoUpdate.getStatus(), saleStageSyncStatusPojoUpdate.getExecuteTime(), null,
                saleStageSyncStatusPojoUpdate.getSyncTimes(), currentExecuteTime, currentExecuteTime, 0);
        // 判断数据库中是否已经有同步记录了，如果有就对原来的记录进行更新
        SaleStageSyncStatusEntity saleStageSyncStatus = saleStageSyncStatusModel.getByStageProcessId(corpid, formId, stageProcessId);
        if (Objects.nonNull(saleStageSyncStatus)) {
            saleStageSyncStatusUpdate.setId(saleStageSyncStatus.getId());
            saleStageSyncStatusUpdate.setAddTime(saleStageSyncStatus.getAddTime());
        }
        saleStageSyncStatusModel.save(saleStageSyncStatusUpdate);

        // 每1000条数据进行分页处理，计算分页页数
        int pageSize = 1000;
        int pageNum = (int) (totalCount % pageSize > 0 ? totalCount / pageSize + 1 : totalCount / pageSize);
        for (int currentPage = 1; currentPage <= pageNum; currentPage++) {
            int startIndex = (currentPage - 1) * pageSize;
            int endIndex = currentPage * pageSize;
            if (Objects.equals(currentPage, pageNum)) {
                endIndex = (int) totalCount;
            }
            LOG.warn("开始同步第{}——{}条数据", startIndex, endIndex);

            List<String> fieldList = Arrays.asList(SalesOpportunityEnum.STAGE_RATIO.getAttr(), SalesOpportunityEnum.STAGE_ID.getAttr());
            EsUtil.setFieldList(sourceBuilder, fieldList);
            PageRequest pageRequest = EsUtil.setPage(sourceBuilder, currentPage, pageSize);
            searchRequest.source(sourceBuilder);
            XbbAggregatedPage<OpportunityEntityExt> esEntities = xbbElasticsearchRestTemplate.queryForPages(pageRequest, searchRequest, OpportunityEntityExt.class);
            if (esEntities == null) {
                throw new XbbException(SystemErrorCodeEnum.API_ERROR_100001);
            }
            List<OpportunityEntityExt> opportunityList = esEntities.getContent();
            Set<Long> currentPageStageIdIn = new HashSet<>();
            Set<Long> currentPageWorkStageIdIn = new HashSet<>();
            List<Long> currentPageStageLogIdIn = new ArrayList<>();
            opportunityList.forEach(opportunity -> {
                Long stageId = opportunity.getData().getLong(SalesOpportunityEnum.STAGE_ID.getAttr());
                if (saleStageMap.isEmpty() || !saleStageMap.containsKey(stageId)) {
                    currentPageStageIdIn.add(stageId);
                }
                if (stageWorkMap.isEmpty() || !stageWorkMap.containsKey(stageId)) {
                    currentPageWorkStageIdIn.add(stageId);
                }
                Long dataId = opportunity.getDataId();
                if (dataStageLogMap.containsKey(dataId)) {
                    currentPageStageLogIdIn.add(dataStageLogMap.get(dataId));
                }
            });

            stageParam.put("idIn", currentPageStageIdIn);
            List<SaleStageEntity> currentPageSaleStageList = saleStageModel.findEntitys(stageParam);
            // 获取本次查询得到的机会阶段数据
            Map<Long, SaleStageEntity> currentPageStageMap = currentPageSaleStageList.stream().collect(Collectors.toMap(SaleStageEntity::getId, Function.identity()));
            // 将本次查询得到的机会阶段数据加入所有机会阶段数据中
            saleStageMap.putAll(currentPageStageMap);

            stageWorkParam.put("stageIdIn", currentPageWorkStageIdIn);
            List<StageWorkEntity> currentPageStageWorkList = stageWorkModel.findEntitys(stageWorkParam);
            Map<Long, List<StageWorkEntity>> currentPageStageWorkMap = currentPageStageWorkList.stream().collect(Collectors.groupingBy(StageWorkEntity::getStageId));
            stageWorkMap.putAll(currentPageStageWorkMap);

            workLogParam.put("stageLogIdIn", currentPageStageLogIdIn);
            List<OpportunityWorkLogEntity> currentPageWorkLogList = opportunityWorkLogModel.findEntitys(workLogParam);
            // 构建阶段记录和任务记录的关系。key-value：阶段记录id-该阶段记录id下的任务记录列表
            Map<Long, List<OpportunityWorkLogEntity>> currentPageStageWorkLogMap = currentPageWorkLogList.stream().collect(Collectors.groupingBy(OpportunityWorkLogEntity::getStageLogId));

            List<OpportunityUpdateDTO> opportunityUpdateList = new ArrayList<>();
            for (OpportunityEntityExt opportunity : opportunityList) {
                Long dataId = opportunity.getDataId();
                Long stageId = opportunity.getData().getLong(SalesOpportunityEnum.STAGE_ID.getAttr());
                SaleStageEntity saleStage = saleStageMap.get(stageId);
                // 是否能够匹配到对应的阶段
                if (Objects.isNull(saleStage)) {
                    continue;
                }
                int opportunityStageRatio = saleStage.getIntoProportion();
                List<StageWorkEntity> stageWorkList = stageWorkMap.get(stageId);
                if (CollectionsUtil.isEmpty(stageWorkList)) {
                    // 没有任务时使用阶段的跳出比例
                    opportunityStageRatio = saleStage.getOutProportion();
                } else {
                    // 获取阶段占比之和
                    int totalWorkProportion = stageWorkList.stream().mapToInt(StageWorkEntity::getProportion).sum();
                    if (totalWorkProportion > 0) {
                        // 有任务且任务占比>0时，使用进入比例+完成的任务占比
                        Long stageLogId = dataStageLogMap.get(dataId);
                        List<OpportunityWorkLogEntity> stageWorkLogList = currentPageStageWorkLogMap.get(stageLogId);
                        if (CollectionsUtil.isNotEmpty(stageWorkLogList)) {
                            // 获取完成的任务id
                            List<Long> finishedWorkIdList = stageWorkLogList.stream().map(OpportunityWorkLogEntity::getStageWorkId).collect(Collectors.toList());
                            // 获取完成的任务占比
                            int totalFinishedWorkProportion = stageWorkList.stream()
                                    .filter(stageWork -> finishedWorkIdList.contains(stageWork.getId()))
                                    .mapToInt(StageWorkEntity::getProportion).sum();
                            opportunityStageRatio = opportunityStageRatio + totalFinishedWorkProportion;
                        }
                    } else {
                        // 有任务但是任务占比=0时，使用使用阶段的跳出比例
                        opportunityStageRatio = saleStage.getOutProportion();
                    }
                }
                int originalStageRatio = opportunity.getData().getIntValue(SalesOpportunityEnum.STAGE_RATIO.getAttr());
                // 新计算的阶段比例跟原来的阶段比例不相等时进行更新
                if (originalStageRatio != opportunityStageRatio) {
                    OpportunityUpdateDTO opportunityUpdateDTO = new OpportunityUpdateDTO();
                    opportunityUpdateDTO.setId(dataId);
                    opportunityUpdateDTO.setCorpid(corpid);
                    JSONObject updateData = new JSONObject();
                    updateData.put(SalesOpportunityEnum.STAGE_RATIO.getAttr(), opportunityStageRatio);
                    opportunityUpdateDTO.setData(updateData);
                    opportunityUpdateList.add(opportunityUpdateDTO);
                }
                LOG.warn("同步之前的阶段比例是{}，同步之后的阶段比例是{}", originalStageRatio, opportunityStageRatio);
            }
            if (CollectionsUtil.isNotEmpty(opportunityUpdateList)) {
                OpportunityUpdateBatchDTO opportunityUpdateBatchDTO = new OpportunityUpdateBatchDTO();
                opportunityUpdateBatchDTO.setCorpid(corpid);
                opportunityUpdateBatchDTO.setOpportunityList(opportunityUpdateList);
                opportunityService.updateBatch4Sync(opportunityUpdateBatchDTO);
            }
        }

        // 同步完成。更新redis【status=同步完成 完成时间=当前时间】
        Long currentFinishTime = DateTimeUtil.getInt();
        saleStageSyncStatusPojoUpdate.setStatus(SaleStageSyncStatusEnum.FINISH.getCode());
        saleStageSyncStatusPojoUpdate.setFinishTime(currentFinishTime);
        paasRedisHelper.setValue(RedisPrefixConstant.SALE_STAGE_SYNC_STATUS, syncStatusKey, saleStageSyncStatusPojoUpdate, TimeConstant.SECONDS_PER_DAY);
        // 更新tb_sale_stage_sync_status表
        saleStageSyncStatusUpdate.setStatus(saleStageSyncStatusPojoUpdate.getStatus());
        saleStageSyncStatusUpdate.setFinishTime(saleStageSyncStatusPojoUpdate.getFinishTime());
        saleStageSyncStatusUpdate.setUpdateTime(currentFinishTime);
        saleStageSyncStatusModel.update(saleStageSyncStatusUpdate);

        // 删除tb_sale_stage_modify_log表中本次已经完成同步的记录
        saleStageModifyLogModel.deleteBatch(saleStageModifyLogVO.getSaleStageModifyLogList(), corpid);

        LOG.warn("公司{}的表单{}的阶段流程{}关联的销售机会同步结束", corpid, formId, stageProcessId);
    }
}
